package com.dtp.core.thread;

import com.dtp.core.support.TaskQueue;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 当所有核心线程都处于繁忙状态时，创建新的线程而不是将任务放入阻塞队列，主要用于io密集型场景
 *
 * @author yanhom
 * @since 1.0.3
 *
 * @remark muse
 **/
public class EagerDtpExecutor extends DtpExecutor {
    private final AtomicInteger submittedTaskCount = new AtomicInteger(0); // 维护已提交但尚未完成的任务数量

    /** -------------------------------------构造函数------------------------------------- */
    public EagerDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                            BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), new AbortPolicy());
    }
    public EagerDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new AbortPolicy());
    }
    public EagerDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                            BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);
    }
    public EagerDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    /** 获取已提交但尚未完成的任务数量 */
    public int getSubmittedTaskCount() {return submittedTaskCount.get();}

    /** 任务后置执行 */
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        submittedTaskCount.decrementAndGet(); // 执行完后将任务数量-1
        super.afterExecute(r, t);             // DtpExecutor.afterExecute(r, t)
    }

    /** 任务执行，如果任务被拒绝并且队列类型是TaskQueue，则依然会强行放入到队列中 */
    @Override
    public void execute(Runnable command) {
        if (command == null) throw new NullPointerException();

        submittedTaskCount.incrementAndGet(); // 任务数+1
        try {
            super.execute(command); // 执行任务DtpExecutor.execute(command)
        } catch (RejectedExecutionException rx) {
            if (getQueue() instanceof TaskQueue) {
                // 如果执行器接近线程池极限，对execute()的并发调用可能（由于使用TaskQueue）导致某些任务被拒绝而不是排队。如果发生这种情况，将它们添加到队列中。
                final TaskQueue queue = (TaskQueue) getQueue(); // 调用的ThreadPoolExecutor#getQueue()
                try {
                    if (!queue.force(command, 0, TimeUnit.MILLISECONDS)) { // 强制向队列中添加任务
                        submittedTaskCount.decrementAndGet(); // 如果添加队列失败，则将任务数量-1
                        throw new RejectedExecutionException("Queue capacity is full.", rx);
                    }
                } catch (InterruptedException x) {
                    submittedTaskCount.decrementAndGet(); // 则将任务数量-1
                    throw new RejectedExecutionException(x);
                }
            } else {
                submittedTaskCount.decrementAndGet(); // 则将任务数量-1
                throw rx;
            }
        }
    }
}
